#include "tcpmanager.h"
#include "Sington.h"

#include <QThread>
#include <QMutexLocker>

volatile qint64 TcpManager::MagicNumber = qrand();


void asyncSocket(QSharedPointer<TransferSocket> socket)
{
    QThread *thread = new QThread;
    socket->moveToThread(thread);
    thread->start();
    QObject::connect(thread, &QThread::finished, thread, &QObject::deleteLater);
    QObject::connect(thread, &QThread::started, socket.get(), [=](){
        qInfo() << "Socket" << socket->socketName() << "run in thread" << (int)socket->thread()->currentThreadId();
    });
}

void registerType()
{
    qRegisterMetaType<QSharedPointer<TransferSocket>>("QSharedPointer<TransferSocket>");
    qRegisterMetaType<QSharedPointer<TcpBlock>>("QSharedPointer<TcpBlock>");
    qRegisterMetaType<TransferType>("TransferType");
    qRegisterMetaType<TcpManager::Functor>("TcpManager::Functor");
}

Acceptor::Acceptor(QObject *parent) :
    QTcpServer(parent)
{
    qDebug() << "Acceptor is running in" << thread()->currentThreadId();
}

Acceptor::~Acceptor()
{
    qDebug() << "Acceptor is destoryed";
}

void Acceptor::incomingConnection(qintptr socketDescriptor)
{
    QSharedPointer<TransferSocket> socket(new TransferSocket);
    if (!socket->setSocketDescriptor(socketDescriptor))
    {
        qDebug() << "Socket Error: " << socket->errorString();
        return;
    }
    QString name = socket->peerAddress().toString() + " " + QString::number(socket->peerPort());
    socket->setSocketName(name);

    qDebug() << "New Socket" << socket->socketName() << "connect to Acceptor.";
    asyncSocket(socket);
    emit newSocket(socket);
}


TcpManager::TcpManager(QObject *parent) :
    QObject(parent),
    m_acceptor(new Acceptor),
    m_thread(nullptr),
    m_rwlock(),
    m_sockets()
{
    registerType();
    connectHandle();
}

TcpManager::~TcpManager()
{
    m_acceptor->close();
    for (auto it = m_sockets.begin(); it != m_sockets.end(); ++it)
    {
        it.value()->disconnectFromHost();
    }

    if (m_thread != nullptr)
    {
        if (m_thread->isRunning())
            m_thread->exit();
        delete m_thread;
    }
    delete m_acceptor;
}

bool TcpManager::listen(const QHostAddress &address, quint16 port)
{
    qInfo() << "TcpServer start listening.";
    return m_acceptor->listen(address, port);
}

void TcpManager::close()
{
    m_acceptor->close();
}

void TcpManager::async()
{
    if (m_thread == nullptr || !m_thread->isRunning())
    {
        if (m_thread == nullptr)
        {
            m_thread = new QThread;
            moveToThread(m_thread);
            m_acceptor->moveToThread(m_thread);
        }
        if (!m_thread->isRunning())
            m_thread->start();

        QObject::connect(m_thread, &QThread::started, this, [=](){
            qDebug() << "Move TcpManager to thread" << (int)this->thread()->currentThreadId();
            qDebug() << "Move Acceptor to thread" << (int)this->thread()->currentThreadId();
        });
    }
}

TcpManager *TcpManager::instance()
{
    return Sington<TcpManager>::getInstance();
}

QList<QSharedPointer<TransferSocket>> TcpManager::getSockets()
{
    QList<QSharedPointer<TransferSocket>> sockets;
    QReadLocker lock(&m_rwlock);
    for (auto it = m_sockets.begin(); it != m_sockets.end(); ++it) {
        sockets.push_back(it.value());
    }
    return sockets;
}

QSharedPointer<TransferSocket> TcpManager::getSocket(const QString& name)
{
    // TODO: 暂时先使用rwlock来实现线程安全
    // 介于在多线程环境下，需要立即返回socket，因此使用锁
    QSharedPointer<TransferSocket> socket;
    QReadLocker lock(&m_rwlock);
    auto it = m_sockets.find(name);
    if (it != m_sockets.end()) {
        socket = it.value();
    }
    return socket;
}

qint64 TcpManager::magicNumber()
{
    return MagicNumber++;
}

void TcpManager::registerFunctor(TransferType type, TcpManager::Functor functor)
{
    if (!runInLoop()) {
        QMetaObject::invokeMethod(this, "registerFunctor",
                                  Q_ARG(TransferType, type),
                                  Q_ARG(TcpManager::Functor, functor));
        return;
    }
    qDebug() << "Register new functor, type is" << type;
    m_functors[type] = functor;
}

void TcpManager::onSocketDisconnect()
{
    TransferSocket* socket = static_cast<TransferSocket*>(sender());
    socket->close();

    qDebug() << "Tcp socket " << socket->socketName() << " disconnect.";

    emit DisConnection(socket->socketName(), socket->peerAddress(), socket->peerPort());
    delSocket(socket->socketName());
    socket = nullptr;
}

void TcpManager::handleNewTcpBlock(QSharedPointer<TcpBlock> block)
{
    auto type = block->type;
    auto functor = m_functors.find(type);
    if (functor != m_functors.end()) {
        (*functor)(block->magicNumber, std::move(block->dataBlock));
    }
}

bool TcpManager::runInLoop() const
{
    return this->thread() == QThread::currentThread();
}

void TcpManager::connectHandle() const
{
    connect(m_acceptor, &Acceptor::newSocket, this, &TcpManager::onNewSocket);
}

void TcpManager::addSocket(QSharedPointer<TransferSocket> socket)
{
    using namespace std::placeholders;
    // TODO: 暂时先使用rwlock来实现线程安全
    // 客户端断开Tcp时调用触发onSocketDisconnect
    connect(socket.get(), &TransferSocket::disconnected, this, &TcpManager::onSocketDisconnect);
    socket->setNewTcpBlockCallBack(std::bind(&TcpManager::handleNewTcpBlock, this, _1));

    QWriteLocker lock(&m_rwlock);
    auto it = m_sockets.find(socket->socketName());
    if (it == m_sockets.end()) {
        m_sockets[socket->socketName()] = socket;
    }
}

void TcpManager::delSocket(const QString& name)
{
    if (!runInLoop()) {
        QMetaObject::invokeMethod(this, "delSocket", Q_ARG(QString, name));
        return;
    }

    QWriteLocker lock(&m_rwlock);
    auto it = m_sockets.find(name);
    (*it)->thread()->exit();
    m_sockets.erase(it);
}

void TcpManager::onNewSocket(QSharedPointer<TransferSocket> socket)
{
    addSocket(socket);
    emit newConnection(socket);
}

void TcpManager::connectToHost(const QString& socketName,
                               const QString& address,
                               const quint16 port)
{
    if (!runInLoop()) {
        QMetaObject::invokeMethod(this, "connectToHost",
                                  Q_ARG(QString, socketName),
                                  Q_ARG(QString, address),
                                  Q_ARG(quint16, port));
        return;
    }
    qDebug() << "Try to connectToHost" << address << ":" << port;
    QSharedPointer<TransferSocket> socket(new TransferSocket);
    socket->setSocketName(socketName);

    // 连接成功
    connect(socket.get(), &TransferSocket::connected, this, [=]() {
        qDebug() << "Connect to host success" << socketName;
        addSocket(socket);
        asyncSocket(socket);
        emit connectToHostSuccess(socket);
    });

    // 连接失败
    connect(socket.get(), QOverload<QAbstractSocket::SocketError>::of(&QAbstractSocket::error),
            this, [=](QAbstractSocket::SocketError socketError) {
        qWarning() << "Connect to host failure" << socketName << " error " << socketError;
        emit connectToHostFailure(address, port);
    });

    socket->connectToHost(address, port);
}

void TcpManager::disconnectFromHost(const QString& socketName)
{
    if (!runInLoop()) {
        QMetaObject::invokeMethod(this, "disconnectFromHost",
                                  Q_ARG(QString, socketName));
        return;
    }

    QWriteLocker lock(&m_rwlock);
    auto it = m_sockets.find(socketName);
    if (it != m_sockets.end()) {
        it.value()->disconnect();
    }
}

void TcpManager::disconnectFromHost(const QString& addr, const quint16 port)
{
    if (!runInLoop()) {
        QMetaObject::invokeMethod(this, "disconnectFromHost",
                                  Q_ARG(QString, addr),
                                  Q_ARG(quint16, port));
        return;
    }
    QHostAddress address(addr);
    decltype(m_sockets) sockets;

    {
    QReadLocker lock(&m_rwlock);
    sockets = m_sockets;
    }

    for (auto it = sockets.begin(); it != sockets.end(); ++it) {
        auto socket = it.value();
        if (socket->peerAddress() == address && socket->peerPort() == port) {
            disconnectFromHost(socket->socketName());
        }
    }
}
